package com.asura.ops.sync.sdk;

import cn.hutool.json.JSONUtil;
import com.asura.base.response.FeignResponseBase;
import com.asura.ops.sync.api.SyncApi;
import com.asura.ops.sync.api.model.CfgClientDto;
import com.asura.ops.sync.api.model.CfgClientMqDto;
import com.asura.ops.sync.sdk.properties.CfgClientProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.cloud.openfeign.EnableFeignClients;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

/**
 * @author: zouyang
 * @date: 2022/8/3
 * @description: 项目启动获取客户端配置
 */
@Configuration
@EnableFeignClients(basePackages = "com.asura.ops.sync")
@Import(CfgClientProperties.class)
@Slf4j
public class CfgClientRunner implements InitializingBean, ApplicationContextAware {

    private SyncApi syncCfgApi;
    private CfgClientProperties cfgClientProperties;

    private ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();

    public CfgClientRunner(SyncApi syncCfgApi, CfgClientProperties cfgClientProperties) {
        this.syncCfgApi = syncCfgApi;
        this.cfgClientProperties = cfgClientProperties;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        ApplicationContextUtil.setApplicationContextAware(applicationContext);
    }

    /**
     * 刷新客户端配置
     */
    private void refreshCfgClient() {
        FeignResponseBase<CfgClientDto> cfgClientDtoRs = syncCfgApi.getCfgClient(cfgClientProperties.getClientCode());
        //第一次获取不到配置kill，后续获取不了，不kill
        if (!cfgClientDtoRs.getResult() && ClientConfigCache.getCfgClientDto() == null) {
            log.info("获取配置失败，退出应用");
            System.exit(100);
        }
        ClientConfigCache.setCfgClientDto(cfgClientDtoRs.getData());
        registerQueueListenerContainer(cfgClientDtoRs.getData());
    }

    /**
     * 监听消息队列
     *
     * @param cfgClientDto
     */
    private void registerQueueListenerContainer(CfgClientDto cfgClientDto) {
        if (!CollectionUtils.isEmpty(cfgClientDto.getCfgClientMqDtoList())) {
            log.info("{}", JSONUtil.toJsonStr(cfgClientDto.getCfgClientMqDtoList()));

            ConnectionFactory connFactory = ApplicationContextUtil.getObject(ConnectionFactory.class, "syncConnectionFactory");
//            JdbcTemplate jdbcTemplate = ApplicationContextUtil.getObject(JdbcTemplate.class);
//            TransactionTemplate transactionTemplate = ApplicationContextUtil.getObject(TransactionTemplate.class);
            List<SimpleMessageListenerContainer> currentListeners = new CopyOnWriteArrayList<>();
            for (CfgClientMqDto mqDto : cfgClientDto.getCfgClientMqDtoList()) {
                String queueName = mqDto.getQueueName();
                if (StringUtils.isEmpty(queueName)) {
                    continue;
                }
                try {
                    List<SimpleMessageListenerContainer> listenerContainers = ClientConfigCache.getListenerContainers();
                    if (!CollectionUtils.isEmpty(listenerContainers)) {
                        for (SimpleMessageListenerContainer listener : listenerContainers) {
                            listener.shutdown();
                            //TODO 处理失败的情况
                        }
                    }
                    //增量
                    SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer();
                    listenerContainer.setConnectionFactory(connFactory);
                    listenerContainer.setConcurrentConsumers(cfgClientProperties.getConcurrentConsumers());
                    listenerContainer.setMaxConcurrentConsumers(cfgClientProperties.getMaxConcurrentConsumers());
                    listenerContainer.setDefaultRequeueRejected(false);
                    listenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);
                    listenerContainer.setMessageListener(new SyncMessageListener());
                    listenerContainer.addQueueNames(queueName);
                    listenerContainer.start();
                    currentListeners.add(listenerContainer);

                    //补偿
                    SimpleMessageListenerContainer listenerContainerAll = new SimpleMessageListenerContainer();
                    listenerContainerAll.setConnectionFactory(connFactory);
                    listenerContainerAll.setConcurrentConsumers(cfgClientProperties.getConcurrentConsumers());
                    listenerContainerAll.setMaxConcurrentConsumers(cfgClientProperties.getMaxConcurrentConsumers());
                    listenerContainerAll.setDefaultRequeueRejected(false);
                    listenerContainerAll.setAcknowledgeMode(AcknowledgeMode.MANUAL);
                    listenerContainerAll.setMessageListener(new SyncMessageListener());
                    listenerContainerAll.addQueueNames(queueName + "_all");
                    listenerContainerAll.start();
                    currentListeners.add(listenerContainerAll);
                } catch (Exception exception) {
                    exception.printStackTrace();
                }
            }
            ClientConfigCache.setListenerContainers(currentListeners);
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        //MQ已有消息，容器还未完全初始化时就消费消息，会报异常：AnnotationConfigServletWebServerApplicationContext has not been refreshed yet
        scheduledExecutorService.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                refreshCfgClient();
            }
        });

        //TODO:不要定时触发，后续界面配置改动进行触发
//        scheduledExecutorService.scheduleAtFixedRate(() -> {
//            try {
//                refreshCfgClient();
//            } catch (Exception exception) {
//                exception.printStackTrace();
//            }
//        }, 0, cfgClientProperties.getRefreshCfgClientSeconds(), TimeUnit.SECONDS);
    }
}
